package com.aispeech.asr.apis.ssc.beans;

import com.aispeech.asr.apis.ssc.config.AIOpsConfig;
import com.aispeech.asr.apis.ssc.config.AppConfig;
import com.aispeech.asr.apis.ssc.config.DopcConfig;
import com.aispeech.asr.apis.ssc.constants.Progress;
import com.aispeech.asr.apis.ssc.data.domain.Align;
import com.aispeech.asr.apis.ssc.data.domain.AudioInfo;
import com.aispeech.asr.apis.ssc.data.domain.SscContext;
import com.aispeech.asr.apis.ssc.data.domain.SscInfoParam;
import com.aispeech.asr.apis.ssc.data.domain.SscResult;
import com.aispeech.asr.apis.ssc.data.domain.TaskDoc;
import com.aispeech.asr.apis.ssc.data.form.Audio;
import com.aispeech.asr.apis.ssc.data.form.Option;
import com.aispeech.asr.clib.arpc.ARpc;
import com.aispeech.asr.clib.arpc.ArpcOutput;
import com.aispeech.asr.clib.arpc.EARpcState;
import com.aispeech.asr.clib.arpc.IARpcCallback;
import com.aispeech.asr.clib.utils.SplittedPcmData;
import com.aispeech.asr.comm.exceptions.AsrException;
import com.aispeech.asr.comm.utils.ShellUtils;
import com.aispeech.jarpc.ArpcException;
import com.aispeech.jarpc.JArpc;
import com.aispeech.jarpc.message.ArpcMessage;
import com.aispeech.jarpc.op.sscvad.SscVadRequest;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.jna.Pointer;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;

@Component
@Slf4j
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class SpeakerSegmentationClustering implements Function<SscContext, SscContext>, IARpcCallback {
    @Autowired
    AppConfig config;
    @Autowired
    AIOpsConfig aiOpsConfig;
    @Autowired
    ObjectMapper objectMapper;
    @Autowired
    DopcConfig DOpsConfig;

    private AsrException exception;
    private SscContext context;

    private Object waiter = new Object();
    //private ARpc rpc;
    private JArpc rpc;
    private String sessionId;

    public void callback(String sessionid, int state, Pointer data, int size, Pointer userData) {
//        log.info("[ onSscCallback ] session: {}, state: {}, size: {}", sessionid, state, size);

        if ((state == EARpcState.S_CONNECTION_BROKEN.getValue()) || (state == EARpcState.S_CONNECTION_CLOSED.getValue())) {
            exception = new AsrException(String.format("[ onSscCallback ] session: %s, state: %d", sessionid, state));
        } else if (state == EARpcState.S_INTERNAL_ERROR.getValue()) {
            byte[] bytes = data == null ? new byte[0] : data.getByteArray(0, size);
            exception = new AsrException(String.format("[ onSscCallback ] session: %s, state: %d, reason: %s",
                    sessionid, state, new String(bytes, StandardCharsets.UTF_8)));
        }
    }

    @SneakyThrows
    @Override
    public SscContext apply(SscContext context) {
        context.addMetric("ssc_begin");

        this.context = context;
        this.exception = null;

        long duration = context.getForm().getDuration();
        Audio audio = context.getForm().getAudio();

        int sampleRate = (audio.getSampleRate() == 8000) ? 8000 : 16000;

        try {
            //rpc = new ARpc();
            //prepareAndStartSsc(sampleRate);
            //doFeedPcmData(duration);
            //doClustering(duration);
            
            doProcess(context, audio);
            
        } finally {
            rpc.close();
            rpc = null;
        }

        context.addMetric("ssc_end");

        return context;
    }
    
    private void doProcess(SscContext context, Audio audioInfo) throws IOException, ArpcException {
    	rpc = new JArpc(DOpsConfig.getHost(),DOpsConfig.getPort(),60*60);
    	
    	//JsonBin jsonBin = new JsonBin(generateParams(taskDoc,audioInfo), Files.readAllBytes(Paths.get(AUDIO)));
    	String resRoot = config.getResRoot();
        String vadRes, sscRes;
        
        Option option = context.getForm().getOption();
        
        if (audioInfo.getSampleRate() == 8000) {
            vadRes = String.format("%s/%s", resRoot, config.getVadRes8k());
            sscRes = String.format("%s/%s", resRoot, config.getSscRes8k());
        } else {
            vadRes = String.format("%s/%s", resRoot, config.getVadRes16k());
            sscRes = String.format("%s/%s", resRoot, config.getSscRes16k());
        }
    	
    	SscVadRequest sscVadRequest = new SscVadRequest(vadRes, sscRes, context.getAudioFile().getAbsolutePath(), option.getnSpeakers(), config.getCudaUnixpath());
        ArpcMessage message = rpc.process(context.getSessionId(), sscVadRequest);
        SscResult result = objectMapper.readValue(message.getBody(), SscResult.class);
        
        context.setSscResult(result);
        
        log.info("[ doProcess.finished ] sessionId: {}",  context.getSessionId());
    	
    }

//    private void doFeedPcmData(long duration) throws IOException, InterruptedException, AsrException {
//        List<Align> aligns = new ArrayList<>();
//        aligns.add(new Align(0, (int)duration, '\0'));
//
//        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
//        objectMapper.writeValue(byteArrayOutputStream, new SscInfoParam(0, aligns));
//        byte[] bytes = byteArrayOutputStream.toByteArray();
//
//        log.info("[ doFeedPcmData ] sessionid: {}, duration: {}, aligns: {}...", sessionId, context.getForm().getDuration(), aligns.size());
//
//        rpc.feed(bytes, bytes.length);
//    }
//
//    private void doClustering(long duration) throws InterruptedException, AsrException {
//        long timeout = ShellUtils.timeoutByDuration(duration);
//
//        log.info("[ doClustering ] waiting... sessionid: {}, duration: {}, timeout: {}", sessionId, duration, timeout);
//
//        ArpcOutput output = rpc.stop();
//        if (exception != null)
//            throw exception;
//
//        if (output.getCode() == EARpcState.S_OK.getValue()) {
////        if (output.getCode() == EARpcState.S_EOF.getValue()) {
//            byte[] data = output.getData();
//            if (data != null && data.length > 0) {
//                try {
//                    SscResult result = objectMapper.readValue(data, 0, data.length, SscResult.class);
//                    int code = result.getCode();
////                  log.info("[ doClustering ] session: {}, code: {}, aligns: {}", sessionid, code, result.getAligns().size());
//
//                    if (code == 0) {
//                        context.setSscResult(result);
//                        log.info("[ doClustering.finished ] sessionId: {}, ssc result code: {}, ", sessionId, code);
//                    } else {
//                        throw new AsrException(String.format("ssc failed. code: %s, data: %s", code, data));
//                    }
//                } catch (Exception e) {
//                    throw new AsrException(String.format("parse ssc result failed. session: %s, bytes: %s, error: %s",
//                            sessionId, data, e.getMessage()));
//                }
//            } else {
//                throw new AsrException(String.format("parse ssc result failed. session: %s, data: %s", sessionId, data));
//            }
//        } else {
//            throw new AsrException(String.format("parse ssc result failed. session: %s, output code: %s", sessionId, output.getCode()));
//        }
//    }
//
//    private void prepareAndStartSsc(int samplerate) {
//        String resRoot = ""; //config.getResRoot();
//        String vadRes, sscRes;
//
//        Option option = context.getForm().getOption();
//
//        if (samplerate == 8000) {
//            vadRes = String.format("%s/%s", resRoot, config.getVadRes8k());
//            sscRes = String.format("%s/%s", resRoot, config.getSscRes8k());
//        } else {
//            vadRes = String.format("%s/%s", resRoot, config.getVadRes16k());
//            sscRes = String.format("%s/%s", resRoot, config.getSscRes16k());
//        }
//
//        String params = String.format("{\"op\":\"ssc-vad\",\"vad_bin_path\":\"%s\",\"ssc_bin_fn\":\"%s\",\"audio_path\":\"%s\",\"speakers\":%d,\"pause_time\":0,\"rpc_path\":\"%s\"}",
//                vadRes,
//                sscRes,
//                context.getAudioFile().getAbsolutePath(),
//                option.getnSpeakers(),
//                config.getCudaUnixpath()
//        );
//
//        rpc.start(aiOpsConfig.getHost(), aiOpsConfig.getPort(), this, params, 3);
//        sessionId = rpc.sessionId();
//
//        log.info("[ doPrepareSSC ] ssc stared. sessionid: {}, params: {}", sessionId, params);
//    }
}
